package transformation;

import Source.AccessSource;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.util.Collector;

/**
 * Create by zhh on 2022/9/5.
 */
public class TransformationApp {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //map(env);
        //filter(env);
        //flatMap(env);
        //KeyBy(env);
        //reduce(env);
        //richMap(env);
        //union(env);
        //connect(env);
        //coMap(env);
        coFlatMap(env);
        env.execute("Transformation App");
    }


    /**
     * ConnectedStreams->DataStream
     * 用于ConnectedStreams在使用flatMap算子的时候用
     */
    public static void coFlatMap(StreamExecutionEnvironment env){
        DataStreamSource<String> source1 = env.fromElements("a b c", "d e f", "g h i");
        DataStreamSource<String> source2 = env.fromElements("1,2,3", "4,5,6", "7,8,9");
        source1.connect(source2).flatMap(new CoFlatMapFunction<String, String, String>() {
            @Override
            public void flatMap1(String value, Collector<String> out) throws Exception {
                String[] splits = value.split(" ");
                for (String split:splits){
                    out.collect(split);
                }
            }

            @Override
            public void flatMap2(String value, Collector<String> out) throws Exception {
                String[] splits = value.split(",");
                for (String split:splits){
                    out.collect(split);
                }
            }

        }).print();
        
    }


    /**
     * ConnectedStreams->DataStream
     * 用于ConnectedStreams在使用Map算子的时候用
     */
    public static void coMap(StreamExecutionEnvironment env){
        DataStreamSource<String> source1 = env.socketTextStream("192.168.32.222", 9527);
        SingleOutputStreamOperator<Integer> source2 = env.socketTextStream("192.168.32.222", 9000)
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String s) throws Exception {
                        return Integer.parseInt(s);
                    }
                });
        //将两个数据源连接
        ConnectedStreams<String, Integer> connect = source1.connect(source2);
        //参数1 第一个source的数据类型   参数2 第二个source的数据类型  参数3 输出的数据流的数据类型
        connect.map(new CoMapFunction<String, Integer, String>() {
            //处理第一个流的业务逻辑
            @Override
            public String map1(String value) throws Exception {
                return value.toUpperCase();
            }
            //处理第二流的业务逻辑
            @Override
            public String map2(Integer value) throws Exception {
                return value*10 +"Source2";
            }
        }).print();
    }


    /**
     * 连接两个数据源，可以是数据类型不一样，连接后是一个ConnectedStreams，无法直接打印，要用map转换
     * union 多流合并  数据结构必须相同  将两个流真正联合在一起
     * connect 双流  数据结构可以不同 两个流连接在一起后可以分开处理 更加灵活
     */
    public static void connect(StreamExecutionEnvironment env){
        DataStreamSource<Access> source1 = env.addSource(new AccessSource());
        DataStreamSource<Access> source2 = env.addSource(new AccessSource());
        SingleOutputStreamOperator<Tuple2<String, Access>> mapedSource2 = source2.map(new MapFunction<Access, Tuple2<String, Access>>() {
            @Override
            public Tuple2<String, Access> map(Access access) throws Exception {
                return Tuple2.of("zh", access);
            }
        });
//        ConnectedStreams<Access, Access> connect = source1.connect(source2);
//        connect.map(new CoMapFunction<Access, Access, Access>() {
//            @Override
//            public Access map1(Access value) throws Exception {
//                return value;
//            }
//
//            @Override
//            public Access map2(Access value) throws Exception {
//                return value;
//            }
//        }).print();
        ConnectedStreams<Access, Tuple2<String, Access>> connect = source1.connect(mapedSource2);
        connect.map(new CoMapFunction<Access, Tuple2<String, Access>, String>() {

            @Override
            public String map1(Access value) throws Exception {
                return value.toString();
            }

            @Override
            public String map2(Tuple2<String, Access> value) throws Exception {
                return value.f0+"->"+value.f1.toString();
            }
        }).print();
    }


    /**
     * 将多个数据类型相同的source联合起来，可以是多个相同的源,联合后是一个DataStream
     */
    public static void union(StreamExecutionEnvironment env){
        DataStreamSource<String> source1 = env.socketTextStream("192.168.32.222", 9527);
        DataStreamSource<String> source2 = env.socketTextStream("192.168.32.222", 9000);

        DataStream<String> union = source1.union(source2);
        source1.union(source1).print();
    }

    /**
     * 增强版map,可以进行生命周期的编写，适用于数据库连接的开启和关闭
     * @param env
     */
    public static void richMap(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.readTextFile("data/access.log");
        SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
            @Override
            //将每行元素变成Access类的数据
            public Access map(String value) throws Exception {
                String[] splits = value.split(",");
                Long time = Long.parseLong(splits[0].trim());
                String domain = splits[1].trim();
                Double traffic = Double.parseDouble(splits[2].trim());
                return new Access(time, domain, traffic);
            }
        });
    }

    /**
     * 把每行数据==>Access
     * 将map算子对应的函数作用到DataStream,产生一个新的DataStream
     * map会作用到已有的DataStream数据集的每一个元素上
     * @param env
     */
    public static void map(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.readTextFile("data/access.log");
        SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
            @Override
            //将每行元素变成Access类的数据
            public Access map(String value) throws Exception {
                String[] splits = value.split(",");
                Long time = Long.parseLong(splits[0].trim());
                String domain = splits[1].trim();
                Double traffic = Double.parseDouble(splits[2].trim());
                return new Access(time, domain, traffic);
            }
        });
        mapStream.print();
    }

    /**
     *过滤traffic>4000
     */
    public static void filter(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.readTextFile("data/access.log");
        SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
            @Override
            //将每行元素变成Access类的数据
            public Access map(String value) throws Exception {
                String[] splits = value.split(",");
                Long time = Long.parseLong(splits[0].trim());
                String domain = splits[1].trim();
                Double traffic = Double.parseDouble(splits[2].trim());
                return new Access(time, domain, traffic);
            }
        });


        SingleOutputStreamOperator<Access> filterStream = mapStream.filter(new FilterFunction<Access>() {
            @Override
            public boolean filter(Access access) throws Exception {
                return access.getTraffic() > 4000;
            }
        });
        filterStream.print();
    }

    /**
     * 用“,”拆开
     * @param env
     */
    public static void flatMap(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.readTextFile("data/access.log");
        //DataStreamSource<String> source = env.socketTextStream("192.168.32.222", 9527);
        source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] splits = value.split(",");
                for (String split:splits){
                    out.collect(split);
                }
            }
        }).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return !"baidu.com".equals(value);
            }
        }).print();


    }

    /**
     * 按照domain分组求和traffic
     * @param env
     */
    public static void KeyBy(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.readTextFile("data/access.log");
        SingleOutputStreamOperator<Access> mapStream = source.map(new MapFunction<String, Access>() {
            @Override
            //将每行元素变成Access类的数据
            public Access map(String value) throws Exception {
                String[] splits = value.split(",");
                Long time = Long.parseLong(splits[0].trim());
                String domain = splits[1].trim();
                Double traffic = Double.parseDouble(splits[2].trim());
                return new Access(time, domain, traffic);
            }
        });
        /**
         * 过时写法
         * mapStream.keyBy("domain").sum("traffic").print();
         */
//        mapStream.keyBy(new KeySelector<Access, String>() {
//            @Override
//            public String getKey(Access access) throws Exception {
//                return access.getDomain();
//            }
//        }).sum("traffic").print();
        mapStream.keyBy(x->x.getDomain())
                .sum("traffic")
                .print();
    }

    /**
     * keyedStream->DataStream
     * wc
     * 1) 读取数据
     * 2) 按照指定分隔符进行拆分
     * 3) 为每个单词赋上一个出现次数为1的值（baidu.com,1）
     * 4) 按照但从进行Keyby
     * 5)分组求和
     *
     * @param env
     */
    public static void reduce(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.readTextFile("data/access.log");
        source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String s, Collector<String> collector) throws Exception {
                String[] splits = s.split(",");
                for (String split:splits){
                    collector.collect(split);
                }
            }
        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return Tuple2.of(s,1);
            }
        }).keyBy(x->x.f0)
                //把相同的值进行计算
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                        return Tuple2.of(value1.f0,value1.f1+value2.f1);
                    }
                })
                .print();

    }
}
